#include <logger.h>

#include "datacenterprivate.h"
#include "datacenter.h"

#include "iot_debug.h"
#include "tools/auto_guard.h"
#include "rapidjson/document.h"
#include "rapidjson/stringbuffer.h"
#include "rapidjson/writer.h"
#include "rapidjson/tools.h"
#include "util_tools.h"

using namespace rapidjson;
using namespace DC;

#define GEN_PUB_TOPIC(rt, app, type, title) \
    do { \
        std::ostringstream ss; \
        ss << app << "/" << type << "/request/datacenter/" << title; \
        rt = ss.str(); \
    } while (0)

#define GEN_SUB_TOPIC(rt, app, type, title) \
    do { \
        std::ostringstream ss; \
        ss << "datacenter/" << type << "/response/" << app << "/" << title; \
        rt = ss.str(); \
    } while (0)

#define GEN_PUB_TOPIC_THIRD(rt, app, type, title) \
    do { \
        std::ostringstream ss; \
        ss << app << "/" << type << "/request/uartManager/" << title; \
        rt = ss.str(); \
    } while (0)

#define GEN_SUB_TOPIC_THIRD(rt, app, type, title) \
    do { \
        std::ostringstream ss; \
        ss << "uartManager/" << type << "/response/" << app << "/" << title; \
        rt = ss.str(); \
    } while (0)

DataCenterPrivate::DataCenterPrivate(DataCenter *p) :
    p_(p),
    topics_(new Topics),
    client_(nullptr)
{

}

DataCenterPrivate::~DataCenterPrivate()
{
    deinit();

    if (topics_)
    {
        delete topics_;
        topics_ = nullptr;
    }

    for (mapMsghandler::iterator it=handlers_.begin(); it!=handlers_.end(); it++) {
        if (it->first) {
            delete [] it->first;
        }
    }
    handlers_.clear();
}

int DataCenterPrivate::init(const char *ip, uint16_t port, const char *user, const char *passwd, const char *clientId)
{
    token_ = 0;
    std::ostringstream os;

    os << "tcp://" << ip << ":" << port;
    int rc =  MQTTAsync_create(&client_, os.str().c_str(), clientId, MQTTCLIENT_PERSISTENCE_NONE, nullptr);
    if (rc != MQTTASYNC_SUCCESS) {
        rc = EXIT_FAILURE;
        goto exit;
    }
    if ((rc = MQTTAsync_setCallbacks(client_, this, nullptr, msgarrvd, nullptr)) != MQTTASYNC_SUCCESS) {
        rc = EXIT_FAILURE;
        goto exit;
    }
    client_id_ = clientId;
    log_info("service: %s  client:%s\n", os.str().c_str(), clientId);
exit:
    if (rc != EXIT_SUCCESS)
        deinit();
    user_ = user;
    passwd_ = passwd;
    if (rc == 0)
    {
        rc = login();
    }

    return rc;
}

void DataCenterPrivate::deinit()
{
    exited_ = true;
    logout();
    if (client_)
    {
        MQTTAsync_destroy(&client_);
        client_ = nullptr;
    }
}

int DataCenterPrivate::login(const char *user_name, const char *passwd, uint16_t keep_alive)
{
    int rc = MQTTASYNC_SUCCESS;
    if (!logined()) {
        MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
        conn_opts.cleansession = 1;
        conn_opts.onSuccess = onConnect;
        conn_opts.onFailure = onConnectFailure;
        conn_opts.context  = this;
        conn_opts.username = user_name;
        conn_opts.password = passwd;
        conn_opts.maxInflight = 10;
        conn_opts.keepAliveInterval = 20;
        keep_alive_ = 20;
        rc = MQTTAsync_connect(client_, &conn_opts);
        if (rc != MQTTASYNC_SUCCESS)
            rc = MQTTASYNC_FAILURE;
        sem_login_.timedwait(10000);
        if (!logined())
            rc = MQTTASYNC_FAILURE;
    }

    log_info("DataCenter login rc:%d\n", rc);
    return rc;
}

int DataCenterPrivate::login(void)
{
    exited_ = false;
    return login(user_.c_str(), passwd_.c_str(), keep_alive_);
}

int DataCenterPrivate::logout()
{
    int rc = MQTTASYNC_SUCCESS;
    if (logined()) {
        MQTTAsync_disconnectOptions disc_opts = MQTTAsync_disconnectOptions_initializer;
        int rc = MQTTAsync_disconnect(client_, &disc_opts);
        if (rc != MQTTASYNC_SUCCESS)
            rc = EXIT_FAILURE;

        log_info("rc:%d\n", rc);
    }

    return rc;
}

bool DataCenterPrivate::logined() const
{
    return (client_ && MQTTAsync_isConnected(client_));
}

int DataCenterPrivate::subTopic(const char *topic, int qos)
{
    if (client_ == nullptr)
        return MQTTASYNC_FAILURE;
    MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
    opts.context = this;
    return MQTTAsync_subscribe(client_, topic, qos, &opts);
}

int DataCenterPrivate::unsubTopic(const char *topic)
{
    if (client_ == nullptr)
        return MQTTASYNC_FAILURE;
    MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
    opts.context = this;
    return MQTTAsync_unsubscribe(client_, topic, &opts);
}

int DataCenterPrivate::pubTopic(const char *topic, const char *msg, int qos)
{
    if (client_ == nullptr)
        return MQTTASYNC_FAILURE;
    MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
    MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
    opts.context = this;
    if (msg) {
        pubmsg.payload = const_cast<char *>(msg);
        pubmsg.payloadlen = static_cast<int>(strlen(msg));
    }
    pubmsg.qos = qos;
    return MQTTAsync_sendMessage(client_, topic, &pubmsg, &opts);
}

bool DataCenterPrivate::registerHandler(const char *topic, PFN_MsgHandler handler)
{
    handlers_[strcpy(new char[strlen(topic)+1], topic)] = handler;
    return true;
}

void DataCenterPrivate::unregisterHandler(const char *topic)
{
    handlers_.erase(topic);
}

bool DataCenterPrivate::registered(const char *topic) const
{
    return handlers_.find(topic) != handlers_.end();
}

void DataCenterPrivate::registerGetModelSchema()
{
    if (topics_->get_modelschema.empty()) {
        GEN_PUB_TOPIC(topics_->get_modelschema, client_id_, "get", "modelschema");
        std::string str;
        GEN_SUB_TOPIC(str, client_id_, "get", "modelschema");
        registerHandler(str.c_str(), reinterpret_cast<PFN_MsgHandler>(&DataCenterPrivate::_onGetModelSchema));
        subTopic(str.c_str());
    }
}

void DataCenterPrivate::registerGetRegister()
{
    if (topics_->get_register.empty()) {
        GEN_PUB_TOPIC(topics_->get_register, clientId(), "get", "register");
        std::string str;
        GEN_SUB_TOPIC(str, clientId(), "get", "register");
        registerHandler(str.c_str(), reinterpret_cast<PFN_MsgHandler>(&DataCenterPrivate::_onGetRegister));
        subTopic(str.c_str());
    }
}

void DataCenterPrivate::registerDeleteModel()
{
    if (topics_->delete_model.empty()) {
        GEN_PUB_TOPIC(topics_->delete_model, clientId(), "action", "deletemodel");
        std::string str;
        GEN_SUB_TOPIC(str, clientId(), "action", "deletemodel");
        registerHandler(str.c_str(), reinterpret_cast<PFN_MsgHandler>(&DataCenterPrivate::_onDelModel));
        subTopic(str.c_str());
    }
}

void DataCenterPrivate::registerGetModel()
{
    if (topics_->get_model.empty()) {
        GEN_PUB_TOPIC(topics_->get_model, clientId(), "get", "model");
        std::string str;
        GEN_SUB_TOPIC(str, clientId(), "get", "model");
        registerHandler(str.c_str(), reinterpret_cast<PFN_MsgHandler>(&DataCenterPrivate::_onGetModel));
        subTopic(str.c_str());
    }
}

void DataCenterPrivate::registerSetModel()
{
    if (topics_->set_model.empty()) {
        GEN_PUB_TOPIC(topics_->set_model, clientId(), "set", "model");
        std::string str;
        GEN_SUB_TOPIC(str, clientId(), "set", "model");
        registerHandler(str.c_str(), reinterpret_cast<PFN_MsgHandler>(&DataCenterPrivate::_onSetModel));
        subTopic(str.c_str());
    }
}

void DataCenterPrivate::registerSetUart()
{
    if (topics_->set_uart.empty()) {
        GEN_PUB_TOPIC_THIRD(topics_->set_uart, clientId(), "set", "commParam");
        std::string str;
        GEN_SUB_TOPIC_THIRD(str, clientId(), "set", "commParam");
        registerHandler(str.c_str(), reinterpret_cast<PFN_MsgHandler>(&DataCenterPrivate::_onSetUart));
        subTopic(str.c_str());
    }
}

void DataCenterPrivate::registerGetDevGUID()
{
    if (topics_->get_dev_guid.empty()) {
        GEN_PUB_TOPIC(topics_->get_dev_guid, clientId(), "get", "guid");
        std::string str;
        GEN_SUB_TOPIC(str, clientId(), "get", "guid");
        registerHandler(str.c_str(), reinterpret_cast<PFN_MsgHandler>(&DataCenterPrivate::_onGetDevGUID));
        subTopic(str.c_str());
    }
}

void DataCenterPrivate::registerRegisterDev()
{
    if (topics_->register_dev.empty()) {
        GEN_PUB_TOPIC(topics_->register_dev, clientId(), "set", "register");
        std::string str;
        GEN_SUB_TOPIC(str, clientId(), "set", "register");
        registerHandler(str.c_str(), reinterpret_cast<PFN_MsgHandler>(&DataCenterPrivate::_onRegisterDev));
        subTopic(str.c_str());
    }
}

void DataCenterPrivate::registerListenDevRegisterNotify()
{
    const char *str = "datacenter/notify/spont/allApp/Updateregister";
    bool regitstered = false;
    for (auto it=handlers_.begin(); it!=handlers_.end(); it++) {
        if (strcmp(it->first, str) == 0) {
            regitstered = true;
            break;
        }
    }
    if (!regitstered) {
        registerHandler(str, reinterpret_cast<PFN_MsgHandler>(&DataCenterPrivate::_onRegisterDevNotify));
    }
}

void DataCenterPrivate::registerUnregisterDev()
{
    if (topics_->unregister_dev.empty()) {
        GEN_PUB_TOPIC(topics_->unregister_dev, clientId(), "action", "unregister");
        std::string str;
        GEN_SUB_TOPIC(str, clientId(), "action", "unregister");
        registerHandler(str.c_str(), reinterpret_cast<PFN_MsgHandler>(&DataCenterPrivate::_onUnregisterDev));
        subTopic(str.c_str());
    }
}

void DataCenterPrivate::registerGetUartInfo()
{
    if (topics_->get_uartinfo.empty()) {
        GEN_PUB_TOPIC_THIRD(topics_->get_uartinfo, client_id_, "get", "uartInfo");
        std::string str;
        GEN_SUB_TOPIC_THIRD(str, client_id_, "get", "uartInfo");
        registerHandler(str.c_str(), reinterpret_cast<PFN_MsgHandler>(&DataCenterPrivate::_onGetUartInfo));
        subTopic(str.c_str());
    }
}

int DataCenterPrivate::msgarrvd(void *context, char *topicName, int topicLen, MQTTAsync_message *message)
{
    DataCenterPrivate *p_this = reinterpret_cast<DataCenterPrivate *>(context);
    if (!message || !topicName) {
        if (message)
            MQTTAsync_freeMessage(&message);
        if (topicName)
            MQTTAsync_free(topicName);
        return true;
    }

    const char *msg = reinterpret_cast<const char *>(message->payload);
    int msg_len = message->payloadlen;
    if (!msg || msg_len==0)
        goto exitCode;
    {
        log_info("%s -> %.*s\n", topicName, msg_len, msg);
        if (p_this->p_->onMsgArrvd(topicName, topicLen, msg, msg_len))
            goto exitCode;
        mapMsghandler::iterator it = p_this->handlers_.find(topicName);
        if ( it == p_this->handlers_.end() ) {
            goto exitCode;
        }
        Document doc;
        doc.Parse(msg, static_cast<size_t>(msg_len));
        (p_this->*it->second)(doc);
    }

exitCode:
    MQTTAsync_freeMessage(&message);
    MQTTAsync_free(topicName);

    return true/*rc*/;
}

void DataCenterPrivate::onConnect(void *context, MQTTAsync_successData */*response*/)
{
    DataCenterPrivate *p_this = reinterpret_cast<DataCenterPrivate *>(context);
    for (mapMsghandler::iterator it=p_this->handlers_.begin(); it!=p_this->handlers_.end(); it++) {
        if (it->first)
            p_this->subTopic(it->first);
    }
    p_this->p_->onConnect();
    p_this->sem_login_.post();
}

void DataCenterPrivate::onConnectFailure(void *context, MQTTAsync_failureData *response)
{
    DataCenterPrivate *p_this = reinterpret_cast<DataCenterPrivate *>(context);
    p_this->p_->onConnectionLost(response->message ? response->message : "connect failure");
}

bool DataCenterPrivate::_onGetModelSchema(const Document &msg)
{
    const Value * body = getChild(msg, "body");
    if (body && body->IsArray()) {
        AUTO_GUARD(gd, MMutex, p_->mtx_);
        for (auto js=body->Begin(); js!=body->End(); js++) {
            if (js->IsString())
                model_names_.push_back(getValueString(js));
        }
        query_model_schema_sem_.post();
    }
    return true;
}

bool DataCenterPrivate::_onGetModel(const Document &msg)
{
    const Value * body = getChild(msg, "body");
    if (body && body->IsArray()) {
        AUTO_GUARD(gd, MMutex, p_->mtx_);
        for (auto js=body->Begin(); js!=body->End(); js++) {
            const char *model_name = getString(js, "model");
            const Value *js_body = getChild(js, "body");
            if (model_name && js_body && js_body->IsArray()) {
                std::string name(model_name);
                models_[name] = ModelItems();
                auto iter = models_.find(name);
                ModelItems & model = iter->second;
                for (auto i=js_body->Begin(); i!=js_body->End(); i++) {
                    if (i->IsObject()) {
                        ModelItem item;
                        item.name = getString(i, "name", "");
                        item.type = getString(i, "type", "");
                        item.unit = getString(i, "unit", "");
                        item.deadzone = getString(i, "deadzone", "0");
                        item.ratio = getString(i, "ratio", "1");
                        item.isReport = getString(i, "isReport", "0");
                        item.userdefine = getString(i, "userdefine", "");
                        model.push_back(item);
                    }
                }
            }
        }
    }
    return true;
}

bool DataCenterPrivate::_onSetModel(const Document &msg)
{
    const char *token = getString(msg, "token");
    const char *status = getString(msg, "status");

    if (token)
    {
        returntoken_ = token;
    }

    if (status)
    {
        returnstatus_ = status;
    }

    log_info("Set Model status: %s.(token:%s)\n", status, token);
    return true;
}

bool DataCenterPrivate::_onSetUart(const Document &msg)
{
    const char *token = getString(msg, "token");
    const char *status = getString(msg, "status");

    if (token)
    {
        returntoken_ = token;
    }

    if (status)
    {
        returnstatus_ = status;
    }

    log_info("Set Uart status: %s.(token:%s)\n", status, token);
    return true;
}

bool DataCenterPrivate::_onDelModel(const Document &msg)
{
    const char *status = getString(msg, "status");
    if (status) {
        if (strcmp(status, "OK") == 0)
            return true;
    }
    return false;
}

bool DataCenterPrivate::_onGetDevGUID(const Document &msg)
{
    const Value * body = getChild(msg, "body");
    if (body && body->IsArray()) {
        return _updateDevGUID(*body);
    }
    return false;
}

bool DataCenterPrivate::_onRegisterDev(const Document &msg)
{
    const char *token = getString(msg, "token");
    const char *status = getString(msg, "status", "unknown");

    if (token)
    {
        returntoken_ = token;
    }

    if (status)
    {
        returnstatus_ = status;
    }

    log_info("register dev status: %s.(token:%s)\n", status, token);

    return true;
}

bool DataCenterPrivate::_onGetRegister(const Document &msg)
{
    const Value * body = getChild(msg, "body");
    if (body && body->IsArray()) {
        AUTO_GUARD(gd, MMutex, p_->mtx_);
        for (auto js=body->Begin(); js!=body->End(); js++) {
            if (!js->IsObject())
                continue;
            const char *model_name = getString(js, "model");
            const Value *js_body = getChild(js, "body");
            const char *port = getString(js, "port");
            if (model_name && port && js_body && js_body->IsArray()) {
                std::string name(model_name);
                auto iter_m=model_devs_.find(name);
                if (iter_m == model_devs_.end()) {
                    model_devs_[name] = PortDevs();
                    iter_m = model_devs_.find(name);
                }
                PortDevs & port_devs = iter_m->second;
                std::string p(port);
                auto iter_p = port_devs.find(p);
                if (iter_p == port_devs.end()) {
                    port_devs[p] = DevInfos();
                    iter_p = port_devs.find(p);
                }
                for (auto i=js_body->Begin(); i!=js_body->End(); i++) {
                    if (i->IsObject()) {
                        DevInfo dev;
                        dev.model = model_name;
                        dev.port  = port;
                        dev.addr  = getString(i, "addr", "");
                        dev.manuID   = getString(i, "manuID", "");
                        dev.manuName = getString(i, "manuName", "");
                        dev.protocolName = getString(i, "ProType", "0");
                        dev.deviceType   = getString(i, "deviceType", "1");
                        dev.nodeID       = getString(i, "nodeID", "0");
                        dev.productID    = getString(i, "productID", "");
                        dev.desc = getString(i, "desc", "");
                        dev.guid = getString(i, "guid", "");
                        dev.dev  = getString(i, "dev", "");
                        iter_p->second.push_back(dev);
                    }
                }
            }
        }
        query_register_sem_.post();
        return true;
    }

    return false;
}

bool DataCenterPrivate::_onUnregisterDev(const Document &msg)
{
    const char *status = getString(msg, "status");
    if (status) {
        if (strcmp(status, "OK") == 0)
            return true;
    }
    return false;
}

bool DataCenterPrivate::_onRegisterDevNotify(const Document &msg)
{
    const Value * body = getChild(msg, "body");
    if (body && body->IsArray())
        return _updateDevGUID(*body);
    return false;
}

bool DataCenterPrivate::_updateDevGUID(const Value &body)
{
    AUTO_GUARD(gd, MMutex, p_->mtx_);
    for (auto js=body.Begin(); js!=body.End(); js++) {
        if (!js->IsObject())
            continue;
        const char *model_name = getString(js, "model");
        const char *port = getString(js, "port");
        const char *addr = getString(js, "addr");
        const char *desc = getString(js, "desc");
        if (model_name && port && addr && desc)
        {
            std::string name(model_name);
            auto iter_m=model_devs_.find(name);
            if (iter_m == model_devs_.end()) {
                model_devs_[name] = PortDevs();
                iter_m = model_devs_.find(name);
            }
            PortDevs & port_devs = iter_m->second;
            auto iter_p = port_devs.find(std::string(port));
            if (iter_p == port_devs.end()) {
                port_devs[port] = DevInfos();
                iter_p = port_devs.find(std::string(port));
            }
            DevInfos & devs = iter_p->second;
            auto iter_dev = devs.begin();
            while (iter_dev != devs.end()) {
                if (iter_dev->addr.compare(addr) == 0 && iter_dev->desc.compare(desc) == 0)
                    break;
                iter_dev++;
            }
            if (iter_dev == devs.end()) {
                DevInfo dev;
                dev.model = name;
                dev.port  = port;
                dev.addr  = addr;
                dev.desc = desc;
                dev.guid = getString(js, "guid", "");
                dev.dev  = getString(js, "dev", "");
                devs.push_back(dev);
            }
            else {
                iter_dev->guid = getString(js, "guid", "");
                iter_dev->dev  = getString(js, "dev", "");
            }
        }
    }
    return true;
}

bool DataCenterPrivate::_onGetUartInfo(const Document &msg)
{
    const char *mode = getString(msg, "mode");
    const char *port = getString(msg, "port");
    const char *appname = NULL;

    if (mode && port)
    {
        uart_infos_.uart_name = port;
        uart_infos_.mode = mode;

        const Value *value = getChild(msg, "status");
        if (value)
        {
            appname = getString(value, "onlyAppName");

            if (appname)
            {
                uart_infos_.appname = appname;
            }
        }
    }

    query_uartinfo_sem_.post();
    return true;
}
